use std::error::Error;
use std::{io, thread};
use std::collections::HashMap;
use std::net::SocketAddr;
use std::process::exit;
use std::sync::Arc;
use std::time::Duration;
use anyhow::{Result};
use clap::Parser;
use dashmap::DashMap;
use log::info;
use tokio;
use tokio::net::{TcpListener, TcpStream};
use tokio::io::AsyncWriteExt;
use simple_redis::{AofType, CmdCache, load_aof, parse_cmd, Redis, RedisConfig, RespEncode, Session, write_aof_always, write_aof_every_sec};

const BUF_SIZE: usize = 4096;

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();

    // 解析参数
    //./simple_redis --port=777   或者   ./simple_redis -p 777
    let redis_config = RedisConfig::parse();
    println!("redis_config is {:?}", redis_config);

    let addr = format!("0.0.0.0:{}", redis_config.port);

    let tcp_listener = TcpListener::bind(addr).await?;

    println!("listener 7379");

    let db = Redis::new(redis_config.clone());

    //
    let sessions: Arc<DashMap<String, Session>> = Arc::new(DashMap::new());


    let cmd_cache = CmdCache::new();


    let aof_session_id = "aof_session".to_string();
    sessions.insert(aof_session_id.clone(), Session::new());

    // 加载aof数据
    let res = load_aof(redis_config.clone().aof_path.clone(), db.clone(), aof_session_id, sessions.clone());


    match res {
        Ok(_) => {
            info!("load aof file success!!!");
        }
        Err(e) => {
            panic!("load aof file error {}", e)
        }
    }


    loop {
        let (stream, addr) = tcp_listener.accept().await?;

        let db_clone = db.clone();
        let session_id = stream.peer_addr().unwrap().to_string();

        println!("session_id is {}", session_id);  //   session_id 就是  客户端的地址


        let sessions_clone = sessions.clone();

        let cmd_cache_clone = cmd_cache.clone();

        sessions_clone.insert(session_id.clone(), Session::new());


        tokio::spawn(async move {
            let res = process_redis_conn(stream, addr, db_clone, session_id, sessions_clone, cmd_cache_clone.clone()).await;
        });


        let redis_config = db.clone().redis_config.clone();
        match redis_config.aof_type {
            // AofType::No => {}
            AofType::Always => {
                // 放在处理命令的函数里面
            }
            AofType::EverySec => {
                let cmd_cache_clone = cmd_cache.clone();
                tokio::spawn(async move {
                    loop {
                        thread::sleep(Duration::from_secs(1));
                        write_aof_every_sec(redis_config.aof_path.clone(), cmd_cache_clone.clone());
                    }
                });
            }
        }
    }
}


async fn process_redis_conn(mut stream: TcpStream
                            , addr: SocketAddr
                            , db: Redis
                            , sesseion_id: String
                            , sessions: Arc<DashMap<String, Session>>
                            , cmd_cache: CmdCache)
                            -> Result<(), Box<dyn Error>> {
    loop {
        stream.readable().await?;

        let mut buf = Vec::with_capacity(BUF_SIZE);

        // let mut msg_list = Vec::new();

        let cmd_cache_c = cmd_cache.clone();
        match stream.try_read_buf(&mut buf) {
            Ok(0) => {
                println!("------------");
                break;
            }
            Ok(n) => {
                let msg = String::from_utf8_lossy(&buf).to_string();

                cmd_cache_c.add_cmd(msg.clone());
                let redis_config = db.clone().redis_config.clone();
                match redis_config.aof_type {
                    // AofType::No => {}
                    AofType::Always => {
                        write_aof_always(redis_config.aof_path.clone(), msg.clone());
                    }
                    AofType::EverySec => {

                        // 放在外面
                    }
                }


                // write_aof(db.redis_config.aof_type.clone(), db.redis_config.aof_path.clone(), msg.clone());

                let res = parse_cmd(msg, db.clone(), sesseion_id.clone(), sessions.clone());
                let temp = res.unwrap().encode();
                let resp = temp.as_slice();

                stream.write_all(resp).await?;
            }
            Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
                continue;
            }
            Err(e) => {
                return Err(e.into());
            }
        }
    }

    println!("close {}  connect ", addr);
    Ok(())
}
